package com.bfxy.disruptor.common.consume;

import java.util.concurrent.atomic.AtomicBoolean;

import com.bfxy.disruptor.common.DataProvider;
import com.bfxy.disruptor.common.event.EventHandler;
import com.bfxy.disruptor.common.LifecycleAware;
import com.bfxy.disruptor.common.SequenceReportingEventHandler;
import com.bfxy.disruptor.common.TimeoutHandler;
import com.bfxy.disruptor.common.balance.Sequence;
import com.bfxy.disruptor.common.balance.SequenceBarrier;
import com.bfxy.disruptor.common.exception.AlertException;
import com.bfxy.disruptor.common.exception.ExceptionHandler;
import com.bfxy.disruptor.common.exception.FatalExceptionHandler;
import com.bfxy.disruptor.common.exception.TimeoutException;
import com.bfxy.disruptor.common.producer.Productor;

//单消费者使用SingleConsumer类，
public final class SingleConsumer<T> implements EventConsumer {// 继承Runnable，所以在线程中运行。

    private final AtomicBoolean running = new AtomicBoolean(false);// 判断是否运行着
    private ExceptionHandler<? super T> exceptionHandler = new FatalExceptionHandler();// 运行过程中遇到异常就处理
    private final DataProvider<T> dataProvider;// 通过序号获取数据，就是RingBuffer。
    // ProcessingSequenceBarrier implements SequenceBarrier
    private final SequenceBarrier sequenceBarrier;// SequenceBarrier栅栏：生产与消费进度协调的类，消费者的栅栏。
    private final EventHandler<? super T> eventHandler;// 消费者处理器
    private final Sequence sequence = new Sequence(Productor.INITIAL_CURSOR_VALUE);// 消费者的序号。每个消费者一个。初始值-1.
    private final TimeoutHandler timeoutHandler;

    public SingleConsumer(final DataProvider<T> dataProvider, final SequenceBarrier sequenceBarrier,
                          final EventHandler<? super T> eventHandler) {
        this.dataProvider = dataProvider;//就是RingBuffer。
        this.sequenceBarrier = sequenceBarrier;
        this.eventHandler = eventHandler;

        if (eventHandler instanceof SequenceReportingEventHandler) {
            ((SequenceReportingEventHandler<?>) eventHandler).setSequenceCallback(sequence);
        }

        timeoutHandler = (eventHandler instanceof TimeoutHandler) ? (TimeoutHandler) eventHandler : null;
    }

    @Override
    public Sequence getSequence() {
        return sequence;
    }

    @Override
    public void halt() {
        running.set(false);
        sequenceBarrier.alert();
    }

    @Override
    public boolean isRunning() {
        return running.get();
    }

    public void setExceptionHandler(final ExceptionHandler<? super T> exceptionHandler) {
        if (null == exceptionHandler) {
            throw new NullPointerException();
        }

        this.exceptionHandler = exceptionHandler;
    }

    /**
     * BatchEventProcessor通过run方法轮询获取实际的数据对象，把数据对象交给消费者去处理的，
     * 交给消费者是利用了消费者的等待策略。
     * 消费者的逻辑，就是在while循环中，不断查询可消费事件，并由用户自定义的消费逻辑eventHandler进行处理。
     * 查询可消费事件的逻辑在SequenceBarrier中。
     */
    @Override
    public void run() {
        if (!running.compareAndSet(false, true))// CAS操作，running是false就置为true，running是true就抛出异常。/避免重复运行
        {
            throw new IllegalStateException("Thread is already running");
        }
        sequenceBarrier.clearAlert();// 序号栅栏清空，清空alerted标志。所有的消费者公用一个sequenceBarrier栅栏。

        notifyStart();// 唤醒线程去工作，调用eventHandler的onStart方法

        T event = null;// 实际的对象

        //  sequence.get()标示当前已经处理的序号,nextSequence表示下一个我要消费的位置。
        long nextSequence = sequence.get() + 1L; 
        try {// 2层while循环
            while (true) {// 死循环，一直等待下一个序号。
                try {
                    //  availableSequence是真实的序号
                    // 消费者消费时候，生产者在生产，就会导致序号不一致。
                	// sequenceBarrier最重要的作用，就是让消费者等待下一个可用的序号
    				// 可用序号可能会大于nextSequence，从而消费者可以一次处理多个事件
    				// 如果该消费者同时也依赖了其他消费者，则会返回最小的那个
                    final long availableSequence = sequenceBarrier.waitFor(nextSequence);// 下次等待14
                    // 如果消费者想要10，但是生产了很多availableSequence=13，一直去消费，直到消费完。
                    while (nextSequence <= availableSequence) {
                        event = dataProvider.get(nextSequence);// RingBuffer取到真实的数据
                        // 真正消费者消费，eventHandler是用户定义的事件消费逻辑
                        eventHandler.onEvent(event, nextSequence, nextSequence == availableSequence);
                        nextSequence++;// 14
                    }
                    // 跟踪自己处理的事件
                    sequence.set(availableSequence);// 13
                }
                // 抛出异常时候，死循环while退出。
                catch (final TimeoutException e) {// 抛出超时异常时while退出
                    notifyTimeout(sequence.get());
                } catch (final AlertException ex) {// 抛出AlertException异常while就退出，捕获到异常就停止return
                    if (!running.get()) {
                        break;
                    }
                } catch (final Throwable ex) {// 抛出异常while退出
                    exceptionHandler.handleEventException(ex, nextSequence, event);
                    sequence.set(nextSequence);
                    nextSequence++;
                }
            }
        } finally {
            notifyShutdown();
            running.set(false);
        }
    }

    private void notifyTimeout(final long availableSequence) {
        try {
            if (timeoutHandler != null) {
                timeoutHandler.onTimeout(availableSequence);
            }
        } catch (Throwable e) {
            exceptionHandler.handleEventException(e, availableSequence, null);
        }
    }

    /**
     * Notifies the EventHandler when this processor is starting up
     */
    private void notifyStart() {
        if (eventHandler instanceof LifecycleAware) {
            try {
                ((LifecycleAware) eventHandler).onStart();
            } catch (final Throwable ex) {
                exceptionHandler.handleOnStartException(ex);
            }
        }
    }

    /**
     * Notifies the EventHandler immediately prior to this processor shutting down
     */
    private void notifyShutdown() {
        if (eventHandler instanceof LifecycleAware) {
            try {
                ((LifecycleAware) eventHandler).onShutdown();
            } catch (final Throwable ex) {
                exceptionHandler.handleOnShutdownException(ex);
            }
        }
    }
}